/*
* Copyright (C) 2010-2012 Magnus Skjegstad
* 
* This program is free software; you can redistribute it and/or
* modify it under the terms of the GNU General Public License
* as published by the Free Software Foundation; either version 2
* of the License, or (at your option) any later version.
*
* This program is distributed in the hope that it will be useful,
* but WITHOUT ANY WARRANTY; without even the implied warranty of
* MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
* GNU General Public License for more details.
*
* You should have received a copy of the GNU General Public License
* along with this program; if not, write to the Free Software
* Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
* 
* The latest version of this library can be downloaded from http://mist-pubsub.googlecode.com
* 
*/
package com.skjegstad.mist2.datatypes;

import com.skjegstad.mist2.exceptions.MistException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.Collection;
import java.util.Map.Entry;

/**
 *
 * @author magnus
 */
public class BloomfilterAcknowledgment implements Acknowledgement {
    private byte[] bloomfilterData;
    private int bloomfilterLengthInBits;
    private int k;
    transient private byte extraBits = 0;
    transient private int hashCount = -1;

    public BloomfilterAcknowledgment(byte[] bloomfilterData, int bloomfilterLengthInBits, int k) throws MistException {
        this.bloomfilterData = bloomfilterData;
        this.bloomfilterLengthInBits = bloomfilterLengthInBits;
        this.k = k;
        if (k < 1)
            throw new MistException("k must be larger than 0. Was " + k);
    }

    public BloomfilterAcknowledgment(Collection<Long> hashes, byte extraBits, int k) throws MistException {
        double bpe = k / Math.log(2);
        int bitLength = (int) (extraBits + Math.ceil(bpe * hashes.size()) + getAccumulatedBitsForK(k));
        int byteLength = (int) Math.ceil(bitLength / 8.0);

        this.extraBits = extraBits; // stored for statistics only
        this.hashCount = hashes.size(); // stored for statistics only

        bloomfilterData = new byte[byteLength];
        bloomfilterLengthInBits = bitLength;
        this.k = k;

        if (k < 1)
            throw new MistException("k must be larger than 0. Was " + k);

        for (long h : hashes) {
            for (int i = 0; i < k; i++)
                this.setAckForK(h, i+1);
            /*int pos = (int) (Math.abs(h) % bitLength);
            int bytePos = pos / 8;
            byte intraBytePos = (byte) (pos % 8);
            bloomfilterData[bytePos] = (byte) (bloomfilterData[bytePos] | (byte)(1 << intraBytePos));*/
        }
    }

    protected static int getAccumulatedBitsForK(int k) {
        int a = (int)Math.round((k / 2.0) * (k-1)); // sum (0+1+..+(k-1)
        return a;
    }

    protected Entry<Integer, Integer> getBoundaries(int k) {
        int totalExtraBitsInclThisK = getAccumulatedBitsForK(k);
        int totalExtraBitsInclPrevK;
        if (k == 1)
            totalExtraBitsInclPrevK = 0;
        else
            totalExtraBitsInclPrevK = getAccumulatedBitsForK(k-1);

        int totalExtraBits = getAccumulatedBitsForK(this.k); 

        int lengthWithoutExtraBits = bloomfilterLengthInBits - totalExtraBits;
        int lengthPerKWithoutExtraBits = lengthWithoutExtraBits / this.k;
        int startOfGivenK = (lengthPerKWithoutExtraBits * (k-1)) + totalExtraBitsInclPrevK;
        int endOfGivenK = (lengthPerKWithoutExtraBits * k) + totalExtraBitsInclThisK;

        return new SimpleEntry<Integer, Integer>(startOfGivenK, endOfGivenK);
    }

    protected int getStartOfK(int k) {
        return this.getBoundaries(k).getKey();
    }

    protected int getEndOfK(int k) {
        return this.getBoundaries(k).getValue();
    }

    protected final void setAckForK(long hash, int k) {
        int startOfK = getStartOfK(k);
        int endOfK = getEndOfK(k);
        int lengthOfK = endOfK - startOfK;

        // TODO fjern debugkode
        if (lengthOfK <= 0) {
            System.err.println("lengthOfK is <= 0 (" + lengthOfK + ") SI FRA TIL MAGNUS");
            System.err.println("startOfK is " + startOfK);
            System.err.println("endOfK is " + endOfK);
            System.err.println("k is " + k);
            System.err.println("bloom filter len (bits) is " + bloomfilterLengthInBits);
            System.err.println("bloom filter len (bytes) is " + bloomfilterData.length);
            System.err.println("extraBit is " + extraBits);
            System.err.println("hash count is " + hashCount);
        }

        int pos = (int) (Math.abs(hash) % lengthOfK) + startOfK;
        int bytePos = pos / 8;
        byte intraBytePos = (byte) (pos % 8);
        bloomfilterData[bytePos] = (byte) (bloomfilterData[bytePos] | (byte)(1 << intraBytePos));
    }

    protected boolean isAckedForK(long hash, int k) {
        
        
        int startOfK = getStartOfK(k);
        int endOfK = getEndOfK(k);
        int lengthOfK = endOfK - startOfK;

        //debug
        //System.out.println("d: lengthOfK:" + lengthOfK + " startOfK:" + startOfK + " endOfK:" + endOfK);
        
        if (lengthOfK == 0) // this can happen if the bloom filter is empty and extrabits == 0
            return false;
        
        int bitPos = (int) (Math.abs(hash) % lengthOfK) + startOfK;
        int bytePos = (int) (bitPos / 8);
        byte intraBytePos = (byte) (bitPos % 8);
        byte mask = (byte) (1 << intraBytePos);
        return ((this.getData()[bytePos] & mask) == mask);
    }

    @Override
    public boolean isAcked(long hash) {
        /*int bitPos = (int) (Math.abs(hash) % this.getLengthInBits());
        int bytePos = (int) (bitPos / 8);
        byte intraBytePos = (byte) (bitPos % 8);
        byte mask = (byte) (1 << intraBytePos);
        return ((this.getData()[bytePos] & mask) == mask);*/
        for (int i = 0; i < k; i++)
            if (!isAckedForK(hash, i+1))
                return false;

        return true;
    }

    @Override
    public byte[] getData() {
        return bloomfilterData;
    }

    @Override
    public int getLengthInBits() {
        return bloomfilterLengthInBits;
    }

    @Override
    public int getK() {
        return k;
    }

    @Override
    public boolean equals(Object obj) {
        if (obj == null) {
            return false;
        }
        if (getClass() != obj.getClass()) {
            return false;
        }
        final BloomfilterAcknowledgment other = (BloomfilterAcknowledgment) obj;
        if (!Arrays.equals(this.bloomfilterData, other.bloomfilterData)) {
            return false;
        }
        if (this.bloomfilterLengthInBits != other.bloomfilterLengthInBits) {
            return false;
        }
        if (this.k != other.k) {
            return false;
        }
        return true;
    }

    @Override
    public int hashCode() {
        int hash = 7;
        hash = 41 * hash + Arrays.hashCode(this.bloomfilterData);
        hash = 41 * hash + this.bloomfilterLengthInBits;
        hash = 41 * hash + this.k;
        return hash;
    }

    
        
}
